package com.xj.hadoop.hive;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

import java.io.IOException;
import java.util.Arrays;

public class HiveInputformat implements InputFormat<LongWritable, BytesWritable>, JobConfigurable {
	TextInputFormat format;
	JobConf job;

	public HiveInputformat() {
		format = new TextInputFormat();
	}

	@Override
	public void configure(JobConf job) {
		this.job = job;
		format.configure(job);
	}

	@Override
	public RecordReader<LongWritable, BytesWritable> getRecordReader(InputSplit genericSplit, JobConf job,
			Reporter reporter) throws IOException {
		reporter.setStatus(genericSplit.toString());
		HiveRecordReader reader = new HiveRecordReader(new LineRecordReader(job, (FileSplit) genericSplit));
		reader.configure(job);
		return reader;
	}

	@Override
	public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
		return format.getSplits(job, numSplits);
	}

	public static class HiveRecordReader implements RecordReader<LongWritable, BytesWritable>, JobConfigurable {
		LineRecordReader reader;
		Text text;

		public HiveRecordReader(LineRecordReader reader) {
			this.reader = reader;
			text = reader.createValue();
		}

		@Override
		public void configure(JobConf arg0) {

		}

		@Override
		public void close() throws IOException {
			reader.close();

		}

		@Override
		public LongWritable createKey() {
			return reader.createKey();
		}

		@Override
		public BytesWritable createValue() {
			return new BytesWritable();
		}

		@Override
		public long getPos() throws IOException {
			return reader.getPos();
		}

		@Override
		public float getProgress() throws IOException {
			return reader.getProgress();
		}

		@Override
		public boolean next(LongWritable key, BytesWritable value) throws IOException {
			while (reader.next(key, text)) {
				System.out.println("inputformat next: " + text.toString());
				/*
				 * String [] strReplace = text.toString().split(" "); for (int i
				 * = 0; i < strReplace.length; i++) { value.set(strReplace[i]);
				 * }
				 */
				byte[] textBytes = text.getBytes();
		        int length = text.getLength();
		        if (length != textBytes.length) {
		            textBytes = Arrays.copyOf(textBytes, length);
		        }
				value.set(textBytes, 0, textBytes.length);
				return true;
			}
			return false;
		}

	}
}
